package com.atguigu.flink.wordcount;

import com.atguigu.flink.pojo.WordCount;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/1/12
 *
      流批一体。
        在某些对数据准确度要求严格的场景。
            白天： 实时计算
            第二天后： 会对昨天产生的数据，再执行批处理，来对数。

            flink可以进行批处理： 使用的是DataSourceAPI
            flink还可以执行流处理: 使用的是 DataStreamSourceAPI

        程序员需要写两套代码！麻烦！
        在1.12版本之后，推出了流批一体。
            使用流式计算的代码，通过简单的参数设置，既可以使用批模式运行，还可以使用流模式运行！
                RuntimeExecutionMode.STREAMING: 流模式。默认
                                       用来计算无界流或有界流。

                 RuntimeExecutionMode.BATCH: 只能用来计算有界流。
                 RuntimeExecutionMode.AUTOMATIC： 如果是有界流，就使用批模式运行！


        Detected an UNBOUNDED source with the 'execution.runtime-mode' set to 'BATCH'.
        This combination is not allowed, please set the 'execution.runtime-mode' to STREAMING or AUTOMATIC

 *
 *
 */
public class Demo8_StreamBatchOne
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //默认情况的运行模式就是流模式
        //所有和env相关的设置，都可以使用代码 | 配置文件 配置
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        env.setParallelism(1);

        DataStreamSource<String> ds = env.readTextFile("data/words.txt");

        SingleOutputStreamOperator<WordCount> ds1 = ds.flatMap(new FlatMapFunction<String, WordCount>()
        {
            @Override
            public void flatMap(String line, Collector<WordCount> out) throws Exception {
                String[] words = line.split(" ");
                for (String word : words) {
                    out.collect(new WordCount(word,1));
                }
            }
        });


        ds1.keyBy(new KeySelector<WordCount, String>()
        {
            @Override
            public String getKey(WordCount data) throws Exception {
                return data.getWord();
            }
        })
          .sum("count")
          .print();


        env.execute();


    }
}
